package com.learn.kafka.simple;

import com.learn.kafka.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;

/**
 * @author: learn.kafka
 */
public class SimpleConsumer extends ConsumerProperties {
    public static void main(String[] args) {
        // 创建当前消费者实例（会自动调用父类构造方法初始化配置）
        SimpleConsumer consumer = new SimpleConsumer();
        consumer.startConsuming();
    }

    // 消费消息的方法
    private void startConsuming() {
        // 从父类获取配置，初始化 Kafka 消费者
        KafkaConsumer<Integer, String> consumer = createConsumer();

        // 订阅主题（可根据需求修改）
        consumer.subscribe(Arrays.asList("mytopic"));

        try {
            while (true) {
                // 拉取消息（超时时间 1 秒）
                ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
                // 遍历消息并打印
                for (ConsumerRecord<Integer, String> record : records) {
                    System.out.printf(
                            "offset = %d, key = %s, value = %s, partition = %d%n",
                            record.offset(), record.key(), record.value(), record.partition()
                    );
                }
            }
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}
